package com.xiaoma.netty.client;

import com.xiaoma.client.RpcClient;
import com.xiaoma.entity.RpcRequest;
import com.xiaoma.entity.RpcResponse;
import com.xiaoma.loadbalancer.RandomLoadBalancer;
import com.xiaoma.registry.NacosServiceDiscovery;
import com.xiaoma.registry.NacosServiceRegistry;
import com.xiaoma.registry.ServiceDiscovery;
import com.xiaoma.registry.ServiceRegistry;
import com.xiaoma.utils.CommonDecoder;
import com.xiaoma.utils.CommonEncoder;
import com.xiaoma.utils.JsonSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicReference;


/**
 * @Author: xiaoma
 * @Date: 2021/7/5 15:31
 * @description: NIO方式消费侧客户端类
 */
public class NettyClient implements RpcClient {

    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    private static final Bootstrap bootstrap;
    private static final EventLoopGroup group;
    private final ServiceDiscovery serviceDiscovery;


    public NettyClient() {
        this.serviceDiscovery = new NacosServiceDiscovery(new RandomLoadBalancer());
    }

    //客户端需要一个事件循环组
    //初始化NioEventLoopGroup
    static {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new CommonDecoder())
                                .addLast(new CommonEncoder(new JsonSerializer()))
                                .addLast(new NettyClientHandler()); //实际处理器
                    }
                });
    }

    @Override
    public Object sendRequest(RpcRequest rpcRequest) {
        //新建一个原子引用结果
        AtomicReference<Object> result = new AtomicReference<>(null);
        try {
            //根据服务名在  nacos 上获取主机地址和端口号
            InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());
            Channel channel = ChannelProvider.get(inetSocketAddress);
            if (!channel.isActive()) {
                group.shutdownGracefully();
                return null;
            }
            //将请求数据写入缓存并刷新 并注册监听器监听
            channel.writeAndFlush(rpcRequest).addListener(future1 -> {
                if (future1.isSuccess()) {
                    logger.info(String.format("客户端发送消息: %s", rpcRequest.toString()));
                } else {
                    logger.error("发送消息时有错误发生: ", future1.cause());
                }
            });
            //对关闭通道进行监听
            channel.closeFuture().sync();
            AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
            //获取服务端响应的RpcResponse 并放入原子引用中  返回
            RpcResponse rpcResponse = channel.attr(key).get();
            result.set(rpcResponse.getData());
        } catch (InterruptedException e) {
            logger.error("发送消息时有错误发生: ", e);
            Thread.currentThread().interrupt();
        }
        return result.get();
    }
}
